package com.shujia.wc;

import java.io.*;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Map {
    public static void main(String[] args) {

        int reduceTasks = 16;

        /**
         * 为每一个文件启动一个线程出力
         *
         */

        ExecutorService executorService = Executors.newFixedThreadPool(4);


        File pfile = new File("java/data/tmp3");
        File[] files = pfile.listFiles();

        for (int i = 0; i < files.length; i++) {

            //为每一个文件启动一个maptask处理
            MapTask mapTask = new MapTask(files[i], reduceTasks);
            executorService.submit(mapTask);

        }

        executorService.shutdown();

    }
}

class MapTask extends Thread {
    private File file;
    private int reduceTasks;

    public MapTask(File file, int reduceTasks) {
        this.file = file;
        this.reduceTasks = reduceTasks;
    }

    @Override
    public void run() {

        try {

            String home = "java/data/tmp4/" + file.getName() + "_";


            /**
             * 每个map为每个reduce生成一个小文件
             *
             */

            //预定于好16个输出流   存到hashmap中   分区编号作为key

            HashMap<Integer, BufferedWriter> writers = new HashMap<>();

            for (int i = 0; i < reduceTasks; i++) {

                BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(home + i));
                writers.put(i, bufferedWriter);

            }


            // 1、读取数据
            FileReader fileReader = new FileReader(file);

            //字符缓冲六
            BufferedReader bufferedReader = new BufferedReader(fileReader);

            String line;
            while ((line = bufferedReader.readLine()) != null) {
                String[] split = line.split(",");
                for (String word : split) {

                    //hash 分区
                    int partition = Math.abs(word.hashCode()) % reduceTasks;


                    //通过分区编号获取输出流
                    BufferedWriter bufferedWriter = writers.get(partition);

                    bufferedWriter.write(word);
                    bufferedWriter.newLine();
                }
            }


            bufferedReader.close();
            fileReader.close();

            for (BufferedWriter bufferedWriter : writers.values()) {
                bufferedWriter.close();
            }


        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
